package com.lzcu.maoyan.mapreduce.service.impl;
/*
    CREATE SCHEMA MAOYANMOVIE;

    CREATE TABLE MAOYANMOVIE.MOVIE_INFO (
        MOVIE_NAME_INFO VARCHAR NOT NULL,
        MOVIE_TIME_INFO TIMESTAMP NOT NULL,
        TICKET_PROPORTION_INFO FLOAT,
        MOVIE_FREQUENCY_INFO FLOAT,
        FREQUENCY_PROPORTION_INFO FLOAT,
        MOVIE_PEOPLE_INFO FLOAT,
        PEOPLE_PROPORTION_INFO FLOAT,
        UP_DAY_INFO INTEGER,
        TOTAL_SESSIONS FLOAT,
        CONSTRAINT PK PRIMARY KEY (MOVIE_NAME_INFO, MOVIE_TIME_INFO)
    );
 */

import java.sql.ResultSet;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;

import com.lzcu.maoyan.mapreduce.connectionPool.ConnectionPhoenix;
import com.lzcu.maoyan.mapreduce.service.PhoenixMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Map;

/**
 * @author WangLiang
 * @date `2024/3/27`
 */
public class PhoenixMapperImpl implements PhoenixMapper {
    private static final Logger logger = LoggerFactory.getLogger(PhoenixMapper.class);

    @Override
    public void insertMovie(Map<String, Object> input) throws Exception {
        if (null != input && input.size() >= 9) {
            String movieName = (String) input.get("movieName");
            String movieTime = (String) input.get("movieTime");
            double ticketProportion = Double.parseDouble((String) input.get("ticketProportion"));
            double movieFrequency = Double.parseDouble((String) input.get("movieFrequency"));
            double frequencyProportion = Double.parseDouble((String) input.get("frequencyProportion"));
            double moviePeople = Double.parseDouble((String) input.get("moviePeople"));
            double peopleProportion = Double.parseDouble((String) input.get("peopleProportion"));
            long upDay = Long.parseLong((String) input.get("upDay"));
            long totalSessions = Long.parseLong((String) input.get("totalSessions"));
            Connection conn = ConnectionPhoenix.getInstance().getConnection();
            String sql = "UPSERT INTO MAOYANMOVIE.MOVIE_INFO (movie_name_info , movie_time_info , ticket_Proportion_info, movie_frequency_info, frequency_Proportion_info,movie_people_info,people_proportion_info,up_day_info,Total_sessions)" +
                    " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
            PreparedStatement pstmt = conn.prepareStatement(sql);
            pstmt.setString(1, movieName);
            pstmt.setString(2, movieTime);
            pstmt.setDouble(3, ticketProportion);
            pstmt.setDouble(4, movieFrequency);
            pstmt.setDouble(5, frequencyProportion);
            pstmt.setDouble(6, moviePeople);
            pstmt.setDouble(7, peopleProportion);
            pstmt.setLong(8, upDay);
            pstmt.setLong(9, totalSessions);
            pstmt.executeUpdate();
            conn.commit();
            pstmt.close();
            conn.close();
        } else {
            assert input != null;
            logger.error(input.toString());
        }
    }

    @Override
    public long getMovieBoxOffice(String movieName, String timeMonth) throws Exception {
        String startTime = timeMonth + "-01 00:00:00.000";
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");

        // 解析字符串为LocalDate（这里假设日期为当月的第一天）
        LocalDate date = LocalDate.parse(timeMonth + "-01", formatter);

        // 月份加一（注意：如果月份是12月，那么年份会加一）
        LocalDate incrementedDate = date.plusMonths(1);

        // 格式化回字符串
        String incrementedDateString = incrementedDate.format(formatter);
        String endTime = incrementedDateString + " 00:00:00.000";
        Connection conn = ConnectionPhoenix.getInstance().getConnection();
        String sql = "SELECT\n" +
                "    MAX(TOTAL_SESSIONS) AS MAX_TOTAL_SESSIONS,\n" +
                "    MIN(TOTAL_SESSIONS) AS MIN_TOTAL_SESSIONS\n" +
                "FROM\n" +
                "    MAOYANMOVIE.MOVIE_INFO\n" +
                "WHERE\n" +
                "\tMOVIE_NAME_INFO = '" + movieName + "' AND \n" +
                "    MOVIE_TIME_INFO >= TO_DATE('" + startTime + "') AND\n" +
                "    MOVIE_TIME_INFO < TO_DATE('" + endTime + "')";
        PreparedStatement pstmt = conn.prepareStatement(sql);
        ResultSet resultSet = pstmt.executeQuery();
        long result = 0;
        if (resultSet.next()) {
            Long max_total_sessions = resultSet.getLong("max_total_sessions");
            Long min_total_sessions = resultSet.getLong("min_total_sessions");
            result = max_total_sessions - min_total_sessions;
            return result;
        }
        pstmt.close();
        conn.close();
        return result;
    }

    @Override
    public long getMovieBoxOffice(String movieName) throws Exception {
        Connection conn = ConnectionPhoenix.getInstance().getConnection();
        String sql = "SELECT MAX(TOTAL_SESSIONS) AS MAX_TOTAL_SESSIONS\n" +
                "FROM MAOYANMOVIE.MOVIE_INFO \n" +
                "WHERE MOVIE_NAME_INFO = '" + movieName + "'";
        PreparedStatement pstmt = conn.prepareStatement(sql);
        ResultSet resultSet = pstmt.executeQuery();
        long result = 0;
        if (resultSet.next()) {
            return resultSet.getLong("MAX_TOTAL_SESSIONS");
        }
        pstmt.close();
        conn.close();
        return result;
    }
}
